package org.leiyang.services;

import com.google.common.util.concurrent.AtomicDouble;
import org.leiyang.entries.BlockFileMetaInfo;
import org.leiyang.loader.RequestIDLoad;
import org.leiyang.loader.SourceDataPreLoad;
import org.roaringbitmap.RoaringBitmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Service;

import java.io.BufferedInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.math.BigDecimal;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.IntStream;

import static org.leiyang.loader.RequestIDLoad.cpu_cores_num;
import static org.leiyang.loader.SourceDataPreLoad.BLOCK_FILE;

/**
 * 应用操作类
 */
@Service
public class BigDataSearcherService implements CommandLineRunner {
    private static final Logger log = LoggerFactory.getLogger(BigDataSearcherService.class);

    /**
     * 根据请求参数文件进行计算
     * @param requeseIdsFilePath  请求参数文件路径
     * @return [0]: org.leiyang.entries.Record#inventory字段的和;
     *          [1]: org.leiyang.entries.Record#price字段的平均值
     */
    public double[] computeByIdFile(String requeseIdsFilePath) {
        if(!SourceDataPreLoad.blockFileAvailable()) {
            throw new RuntimeException("源数据正在更新，请稍后再试！");
        }

        int[] allRequestIds = null;
        try {
            allRequestIds = RequestIDLoad.loadRequestIDs(requeseIdsFilePath);
        }catch (IOException ie) {
            log.error("请求参数文件不存在!");
            return new double[]{0D, 0D};
        }

        if(Objects.isNull(allRequestIds)) {
            throw new RuntimeException("读取源数据文件失败！");
        }

        List<BlockFileMetaInfo> blockFileMetaInfos = SourceDataPreLoad.blockFilesMetaInfoGot();
        blockFileMetaInfos.sort(Comparator.comparingInt(BlockFileMetaInfo::getMinId));
        Map<BlockFileMetaInfo, int[]> requestIdsMapToBlockFileMap = RequestIDLoad.requestIdsMapToBlockFile(
                blockFileMetaInfos.toArray(new BlockFileMetaInfo[0]),
                allRequestIds
        );

        final LongAdder sumInventory = new LongAdder(); //inventory的求和
        final AtomicDouble sumPrice = new AtomicDouble();   //price的求和


        ForkJoinPool customPool = new ForkJoinPool(cpu_cores_num * 2);
        Future<?> future = customPool.submit(() -> requestIdsMapToBlockFileMap.entrySet().stream().parallel().forEach(blockFileMetaInfoEntry ->
                computerByIdsFromBlockFileMetaInfo(blockFileMetaInfoEntry.getKey(), blockFileMetaInfoEntry.getValue(), sumInventory, sumPrice)
        ));
        customPool.shutdown();
        try {
            future.get(); //阻塞等待
        } catch (InterruptedException | ExecutionException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Parallel computerByIdsFromBlockFileMetaInfo failed", e);
        }

        BigDecimal avgPrice = BigDecimal.valueOf(sumPrice.get()).divide(BigDecimal.valueOf(allRequestIds.length), 2, BigDecimal.ROUND_UP);
        return new double[]{sumInventory.sum(), avgPrice.doubleValue()};
    }

    /**
     * 根据请求参数文件进行计算 (使用RoaringBitmap替代int[]的改进版)
     * @param requeseIdsFilePath  请求参数文件路径
     * @return [0]: org.leiyang.entries.Record#inventory字段的和;
     *          [1]: org.leiyang.entries.Record#price字段的平均值
     */
    public double[] computeByIdFileV1(String requeseIdsFilePath) {
        if(!SourceDataPreLoad.blockFileAvailable()) {
            throw new RuntimeException("源数据正在更新，请稍后再试！");
        }

        RoaringBitmap allRequestIds = RequestIDLoad.loadRequestIDsV1(requeseIdsFilePath);
        if(Objects.isNull(allRequestIds)) {
            throw new RuntimeException("读取源数据文件失败！");
        }

        List<BlockFileMetaInfo> blockFileMetaInfos = SourceDataPreLoad.blockFilesMetaInfoGot();
        blockFileMetaInfos.sort(Comparator.comparingInt(BlockFileMetaInfo::getMinId));
        Map<BlockFileMetaInfo, RoaringBitmap> requestIdsMapToBlockFileMap = RequestIDLoad.requestIdsMapToBlockFileV1(
                blockFileMetaInfos.toArray(new BlockFileMetaInfo[0]),
                allRequestIds
        );

        final LongAdder sumInventory = new LongAdder(); //inventory的求和
        final AtomicDouble sumPrice = new AtomicDouble();   //price的求和


        ForkJoinPool customPool = new ForkJoinPool(cpu_cores_num * 2);
        Future<?> future = customPool.submit(() -> requestIdsMapToBlockFileMap.entrySet().stream().parallel().forEach(blockFileMetaInfoEntry ->
                computerByIdsFromBlockFileMetaInfoV1(blockFileMetaInfoEntry.getKey(), blockFileMetaInfoEntry.getValue(), sumInventory, sumPrice)
        ));
        customPool.shutdown();
        try {
            future.get(); //阻塞等待
        } catch (InterruptedException | ExecutionException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Parallel computerByIdsFromBlockFileMetaInfo failed", e);
        }

        BigDecimal avgPrice = BigDecimal.valueOf(sumPrice.get()).divide(BigDecimal.valueOf(allRequestIds.getCardinality()), 2, BigDecimal.ROUND_UP);
        return new double[]{sumInventory.sum(), avgPrice.doubleValue()};
    }

    /**
     * 重置源数据
     */
    public void resetSourceData(){
        SourceDataPreLoad.reSetSourceData();
    }


    /**
     * 单个BlockFileMetaInfo下计算多个id
     * @param blockFileMetaInfo block小文件的元信息
     * @param ids 多个id,id的个数不会超过1000000
     */
    private void computerByIdsFromBlockFileMetaInfo(BlockFileMetaInfo blockFileMetaInfo, int[] ids, LongAdder sumInventory, AtomicDouble sumPrice) {
        final int idsCount = ids.length;
        int[] inventoryArray = new int[1000000];
        float[] priceArray = new float[1000000];
        //存储blockFileMetaInfo里所有id和其index的映射关系
        Map<Integer, Integer> idAndIndexMap = new ConcurrentHashMap<>(1000000);

        try {
            Path blockFile = Paths.get(BLOCK_FILE, blockFileMetaInfo.getFileName());
            ObjectInputStream ois = new ObjectInputStream(
                    new BufferedInputStream(Files.newInputStream(blockFile)));
            boolean isEnd = false;
            int idIndex = 0;
            while (!isEnd) {
                try {
                    int id = ois.readInt();
                    int inventory = ois.readInt();
                    float price = ois.readFloat();
//                    log.info("读取到内容: {},{},{}", id, inventory, price);
                    idAndIndexMap.put(id, idIndex);
                    //缓存对应的inventory和price
                    inventoryArray[idIndex] = inventory;
                    priceArray[idIndex] = price;
                    idIndex++;
                }catch (EOFException e) {
                    log.debug("已读取完BLOCK小文件: {}", blockFileMetaInfo.getFileName());
                    isEnd = true;
                }
            }

            ForkJoinPool customPool = new ForkJoinPool(4);
            customPool.submit(() ->
                    IntStream.range(0, idsCount).parallel().forEach(i -> {
                        int requestId = ids[i];
                        int oriIdIndex = idAndIndexMap.getOrDefault(requestId, -1);
//                        log.info("{}对应的block里id的index: {}", requestId, oriIdIndex);
                        if(oriIdIndex > 0){
                            sumInventory.add(inventoryArray[oriIdIndex]);
                            sumPrice.addAndGet(priceArray[oriIdIndex]);
                        }
                    })
            ).join();
            customPool.shutdown();

            log.info("获取到的sumInventory的总和: {}, sumPrice的总和: {}", sumInventory.sum(), sumPrice.get());

        } catch (Exception e){
            log.error("对BLOCK小文件:{}进行筛选id计算操作异常", blockFileMetaInfo.getFileName(), e);
        }
    }

    /**
     * 单个BlockFileMetaInfo下计算多个id
     * org.leiyang.services.BigDataSearcherService#computerByIdsFromBlockFileMetaInfo的改进版
     * @param blockFileMetaInfo block小文件的元信息
     * @param ids 多个id,id的个数不会超过1000000
     */
    private void computerByIdsFromBlockFileMetaInfoV1(BlockFileMetaInfo blockFileMetaInfo, RoaringBitmap ids, LongAdder sumInventory, AtomicDouble sumPrice) {
        int[] inventoryArray = new int[1000000];
        float[] priceArray = new float[1000000];
        //存储blockFileMetaInfo里所有id和其index的映射关系
        Map<Integer, Integer> idAndIndexMap = new ConcurrentHashMap<>(1000000);

        try {
            Path blockFile = Paths.get(BLOCK_FILE, blockFileMetaInfo.getFileName());
            ObjectInputStream ois = new ObjectInputStream(
                    new BufferedInputStream(Files.newInputStream(blockFile)));
            boolean isEnd = false;
            int idIndex = 0;
            while (!isEnd) {
                try {
                    int id = ois.readInt();
                    int inventory = ois.readInt();
                    float price = ois.readFloat();
//                    log.info("读取到内容: {},{},{}", id, inventory, price);
                    idAndIndexMap.put(id, idIndex);
                    //缓存对应的inventory和price
                    inventoryArray[idIndex] = inventory;
                    priceArray[idIndex] = price;
                    idIndex++;
                }catch (EOFException e) {
                    log.debug("已读取完BLOCK小文件: {}", blockFileMetaInfo.getFileName());
                    isEnd = true;
                }
            }

            ForkJoinPool customPool = new ForkJoinPool(4);
            customPool.submit(() ->
                ids.stream().parallel().forEach(requestId -> {
                    int oriIdIndex = idAndIndexMap.getOrDefault(requestId, -1);
//                        log.info("{}对应的block里id的index: {}", requestId, oriIdIndex);
                    if(oriIdIndex > 0){
                        sumInventory.add(inventoryArray[oriIdIndex]);
                        sumPrice.addAndGet(priceArray[oriIdIndex]);
                    }
                })
            ).join();
            customPool.shutdown();
            log.info("获取到的sumInventory的总和: {}, sumPrice的总和: {}", sumInventory.sum(), sumPrice.get());

        } catch (Exception e){
            log.error("对BLOCK小文件:{}进行筛选id计算操作异常", blockFileMetaInfo.getFileName(), e);
        }
    }


    @Override
    public void run(String... args) throws Exception {
        log.info("开始对源数据进行预热操作..");
        SourceDataPreLoad.preLoadSourceData();

    }
}
